package com.study.connect.client.impl;

import com.study.common.utils.NetAddressUtil;
import com.study.connect.client.IConnectService;
import com.study.connect.client.handler.ClientHandler;
import com.study.connect.client.handler.HandlerCallable;
import com.study.connect.exception.MyRpcConnectionCloseException;
import com.study.connect.exception.MyRpcTimeOutException;
import com.study.connect.handler.ByteToInvokeResultDecoderHandler;
import com.study.connect.handler.ServiceInvokeToByteEncoderHandler;
import com.study.connect.message.EmpResult;
import com.study.connect.message.RpcExceptionResult;
import com.study.connect.message.ServiceInvokeDTO;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.LengthFieldPrepender;
import io.netty.handler.timeout.IdleStateHandler;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.util.Map;
import java.util.concurrent.*;

/**
 * @包名称 com.study.connect.client.impl
 * @类描述 <pre></pre>
 * @作者 zhangbo
 * @创建时间 2020/6/13.
 * @版本 1.00
 * @修改记录 <pre>
 *     版本           修改人         修改日期         修改内容描述
 *     -----------------------------------------------------------
 *     1.00           zhangbo        2020/6/13.           类创建
 *     -----------------------------------------------------------
 * </pre>
 */
@Slf4j
@Component
public class ConnectServiceImpl implements IConnectService {
    /*连接对象池*/
    public static final ThreadLocal<Map<String, ChannelFuture>> CHANNEL_FUTURE_LOCAL = new ThreadLocal<>();

    @Value("${RPC.TIME.OUT:3000}")
    private int timeout;

    public static final ThreadLocal<ClientHandler> CLIENT_HANDLER_LOCAL = new ThreadLocal<>();

    private static final ThreadPoolExecutor THREAD_POOL_EXECUTOR = new ThreadPoolExecutor(20, 200, 10, TimeUnit.MINUTES, new ArrayBlockingQueue<Runnable>(200), new ThreadFactory() {
        @Override
        public Thread newThread(Runnable r) {
            Thread thread = new Thread(r);
            thread.setName("CONNECT-THREAD-POOL");
            return thread;
        }
    });

    @Override
    public ChannelFuture getConnectService(String serviceIp, Integer servicePort) {
        StringBuffer key = new StringBuffer();
        key.append("service:").append(serviceIp).append("-port").append(servicePort);
        Map<String, ChannelFuture> channelFutureMap = CHANNEL_FUTURE_LOCAL.get();
        if (null == channelFutureMap || channelFutureMap.isEmpty()) {
            synchronized (CHANNEL_FUTURE_LOCAL) {
                /*构建连接对象*/
                channelFutureMap = CHANNEL_FUTURE_LOCAL.get();
                if (null == channelFutureMap || channelFutureMap.isEmpty()) {
                    /*二次校验*/
                    CHANNEL_FUTURE_LOCAL.set(new ConcurrentHashMap<>(16));
                }
            }
        }
        channelFutureMap = CHANNEL_FUTURE_LOCAL.get();
        if (!channelFutureMap.containsKey(key.toString())) {
            /*如果不包含构建的连接，则创建一个构建的新连接*/
            synchronized (CHANNEL_FUTURE_LOCAL) {
                if (!channelFutureMap.containsKey(key.toString())) {
                    /*二次校验*/
                    channelFutureMap.put(key.toString(), createChannelFutrue(serviceIp, servicePort));
                }
            }
        }

        return channelFutureMap.get(key.toString());
    }


    public void removeConnectService(String serviceIp, Integer servicePort) {
        StringBuffer key = new StringBuffer();
        key.append("service:").append(serviceIp).append("-port").append(servicePort);

        Map<String, ChannelFuture> channelFutureMap = CHANNEL_FUTURE_LOCAL.get();
        if (null != channelFutureMap && !channelFutureMap.isEmpty() && channelFutureMap.containsKey(key.toString())) {
                /*构建连接对象*/
            channelFutureMap.remove(key.toString());
        }
    }


    @Override
    public Object serviceInvoke(ServiceInvokeDTO serviceInvokeDTO) throws Exception {
        ChannelFuture connectService = getConnectService(serviceInvokeDTO.getServiceIp(), serviceInvokeDTO.getServicePort());
        HandlerCallable handlerCallable = new HandlerCallable(connectService, serviceInvokeDTO, timeout);
        try {
            Future futrue = THREAD_POOL_EXECUTOR.submit(handlerCallable);
            Object result = futrue.get();
            if (result instanceof EmpResult) {
                return null;
            } else if (result instanceof RpcExceptionResult) {
                throw ((RpcExceptionResult) result).getException();
            }
            log.debug("服务器返回参数：{}", result);
            return result;
        } catch (ExecutionException e) {
            /*当前连接失败了，需要剔除，并进行关闭*/
            if (e.getCause() instanceof MyRpcConnectionCloseException) {
                log.error("连接节点已关闭，剔除节点：{}", serviceInvokeDTO);
                removeAndCloseChannelFuture(serviceInvokeDTO);
                throw new MyRpcConnectionCloseException(e.getCause().getMessage());
            }
            if (e.getCause() instanceof MyRpcTimeOutException) {
                log.error("连接节点已关闭，剔除节点：{}", serviceInvokeDTO);
                removeAndCloseChannelFuture(serviceInvokeDTO);
                throw new MyRpcTimeOutException(e.getCause().getMessage());
            }
            throw e;
        } catch (Exception e) {
            log.error("远程调用异常：{}", e);
        }
        throw new RuntimeException("服务远程调用异常");
    }

    public void removeAndCloseChannelFuture(ServiceInvokeDTO serviceInvokeDTO) {
        try {
            ChannelFuture connectService = getConnectService(serviceInvokeDTO.getServiceIp(), serviceInvokeDTO.getServicePort());
            removeConnectService(serviceInvokeDTO.getServiceIp(), serviceInvokeDTO.getServicePort());
            connectService.channel().closeFuture().sync();
        } catch (Exception e) {
            log.error("连接已经废弃，关闭连接异常：{}", e);
        }
    }


    public ChannelFuture createChannelFutrue(String serviceIp, Integer servicePort) {
        EventLoopGroup clientGroup = new NioEventLoopGroup();
        Bootstrap bootstrap = new Bootstrap();
        final ClientHandler clientHandler = new ClientHandler();
        CLIENT_HANDLER_LOCAL.set(clientHandler);
        final int port = NetAddressUtil.getPort();
        log.info("获取本地绑定端口号：{}", port);
        bootstrap.group(clientGroup).localAddress(port).option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000)
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        ChannelPipeline pipeline = socketChannel.pipeline();
                        pipeline.addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4))
                                    /*LengthFieldPrepender参数含义：前置长度字段所占的字节数*/
                                .addLast(new LengthFieldPrepender(4))
                                /*心跳检测机制，参数：读心跳、写心跳、读写心跳，时间单位*/
                                .addLast(new IdleStateHandler(5, 0, 0, TimeUnit.SECONDS))
                                .addLast(new ServiceInvokeToByteEncoderHandler())
                                .addLast(new ByteToInvokeResultDecoderHandler())
                                .addLast(clientHandler);
                    }
                });

        try {
            ChannelFuture future = bootstrap.connect(serviceIp, servicePort).sync();
            return future;
        } catch (InterruptedException e) {
            log.error("服务连接异常：{}", e);
            Thread.currentThread().interrupt();
        }
        return null;
    }
}
